package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.PaperScoreSplitBean;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import groupOne.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * drop table if exists dws_test_paper_score_split_window;
 * create table if not exists dws_test_paper_score_split_window(
 *     stt datetime,
 *     edt datetime,
 *     cur_date date,
 *     paper_id varchar(20),
 *     paper_name varchar(200),
 *     score_level varchar(20),
 *     uu_ct bigint replace
 * )
 * engine = olap
 * aggregate key(stt,edt,cur_date,paper_id,paper_name,score_level)
 * comment "考试域试卷成绩分布汇总表"
 * partition by range(cur_date)()
 * distributed by hash(stt) buckets 10 properties(
 *   "replication_num" = "3",
 *   "dynamic_partition.enable" = "true",
 *   "dynamic_partition.time_unit" = "DAY",
 *   "dynamic_partition.start" = "-1",
 *   "dynamic_partition.end" = "1",
 *   "dynamic_partition.prefix" = "par",
 *   "dynamic_partition.buckets" = "10",
 *   "dynamic_partition.hot_partition_num" = "3"
 * );
 */
public class Dws_02_TestPaperScoreSplitXiang extends BaseAppOneStream {
    public static void main(String[] args) {
        new Dws_02_TestPaperScoreSplitXiang().init(
                6005,
                2,
                "Dws_02_TestPaperScoreSplit",
                Constant.TOPIC_DWD_TEST_TEST_EXAM
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //封装pojo
        SingleOutputStreamOperator<PaperScoreSplitBean> pojoStream = transFunc(stream);
        //聚合
        SingleOutputStreamOperator<PaperScoreSplitBean> aggStream = aggFunc(pojoStream);
        //dim
        SingleOutputStreamOperator<PaperScoreSplitBean> resultStream = addDim(aggStream);
        //写入doris
        writeDoris(resultStream);
    }

    private void writeDoris(SingleOutputStreamOperator<PaperScoreSplitBean> resultStream) {
        resultStream
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy= PropertyNamingStrategy.SnakeCase;
                    return JSON.toJSONString(bean,config);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_test_paper_score_split_window"));
    }

    private SingleOutputStreamOperator<PaperScoreSplitBean> addDim(SingleOutputStreamOperator<PaperScoreSplitBean> aggStream) {
        return AsyncDataStream.unorderedWait(
                aggStream,
                new DimAsyncFunction<PaperScoreSplitBean>() {
                    @Override
                    public String getTable() {
                        return "dim_test_paper";
                    }

                    @Override
                    public String getId(PaperScoreSplitBean input) {
                        return input.getPaperId();
                    }

                    @Override
                    public void addDim(JSONObject dim, PaperScoreSplitBean bean) {
                        bean.setPaperName(dim.getString("PAPER_TITLE"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
    }

    private SingleOutputStreamOperator<PaperScoreSplitBean> aggFunc(SingleOutputStreamOperator<PaperScoreSplitBean> pojoStream) {
        return pojoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<PaperScoreSplitBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((t,l) -> t.getTs())
                )
                .keyBy(bean -> bean.getPaperId()+":"+bean.getScoreLevel())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<PaperScoreSplitBean>() {
                            @Override
                            public PaperScoreSplitBean reduce(PaperScoreSplitBean bean1,
                                                              PaperScoreSplitBean bean2) throws Exception {
                                bean1.setUuCt(bean1.getUuCt() + bean2.getUuCt());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<PaperScoreSplitBean, PaperScoreSplitBean, String, TimeWindow>() {
                            @Override
                            public void process(String pisl,
                                                ProcessWindowFunction<PaperScoreSplitBean, PaperScoreSplitBean, String, TimeWindow>.Context ctx,
                                                Iterable<PaperScoreSplitBean> iterable,
                                                Collector<PaperScoreSplitBean> collector) throws Exception {
                                PaperScoreSplitBean bean = iterable.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDate(ctx.window().getStart()+1000));

                                collector.collect(bean);
                            }
                        });
    }

    private SingleOutputStreamOperator<PaperScoreSplitBean> transFunc(DataStreamSource<String> stream) {
        return stream
                .map(new MapFunction<String, PaperScoreSplitBean>() {
                    @Override
                    public PaperScoreSplitBean map(String json) throws Exception {
                        JSONObject obj = JSON.parseObject(json);
                        String paperId = obj.getString("paper_id");
                        Double score = obj.getDouble("score");
                        Long ts = obj.getLong("ts")*1000;

                        PaperScoreSplitBean bean = new PaperScoreSplitBean(
                                "","","",
                                paperId,"","",1l,
                                ts
                                );

                        if (score >= 90){
                            bean.setScoreLevel("优秀");
                        }else if ( score >= 80){
                            bean.setScoreLevel("良好");
                        }else if (score >= 60){
                            bean.setScoreLevel("及格");
                        }else {
                            bean.setScoreLevel("不及格");
                        }

                        return bean;
                    }
                });
    }
}
